package com.gxlevi.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class PartitionProducer {
    /**
     * kafka生产数据,通过不同的方式,将数据写入到不同的分区里面去
     *
     * @param args
     */
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "node01:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        //配置自定义的分区类
//        props.put("partitioner.class", "com.gxlevi.kafka.producer.MyPartitioner");
        //获取kafkaProducer这个类
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props);
        for (int i = 0; i < 100; i++) {
            //第一种策略:既没有指定分区号,也没有指定数据key,那么就会使用轮询的方式将数据均匀的发送到不同的分区里面去
            ProducerRecord<String, String> producerRecord1 = new ProducerRecord<>("mypartition", "message" + i);
            //第二种策略:如果没有指定分区号,指定了数据key,通过key.hashCode%numPartitions来计算数据究竟会保存在哪个分区里面
            //注意,如果key没有变化,key%myPartition = 固定值,把数据全部写到某一个分区里面去
            ProducerRecord<String, String> producerRecord2 = new ProducerRecord<>("mypartition", "mykey", "message" + i);
            //第三种策略:如果指定分区号,可以将数据直接写入到对应的分区里面去
            ProducerRecord<String, String> producerRecord3 = new ProducerRecord<String, String>("mypartition", 0, "mykey", "message" + i);
            //自定义策略:如果不自定义分区的规则,会将数据使用轮询的方式均匀的发送到各个分区里面去
            ProducerRecord<String, String> producerRecord4 = new ProducerRecord<>("mypartition", "message" + i);
            kafkaProducer.send(producerRecord4);
        }
        kafkaProducer.close();
    }
}
